package com.qf.pubsub;

import com.qf.util.MQConnUtil;
import com.rabbitmq.client.*;
import org.junit.Test;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class Consumer {

    @Test
    public void consume1() throws Exception, TimeoutException {
        //1. c
        Connection conn = MQConnUtil.getConn();

        //2. c
        Channel channel = conn.createChannel();

        //3. 创建exchange     DIRECT - 直接和Queue进行绑定即可,routingKey写空串
        channel.exchangeDeclare(Publisher.EXCHANGE_NAME, BuiltinExchangeType.DIRECT);

        //4. 创建queue
        channel.queueDeclare(Publisher.QUEUE_NAME1,true,false,false,null);
        channel.queueDeclare(Publisher.QUEUE_NAME2,true,false,false,null);

        //5. 将exchange和队列绑定到一起
        channel.queueBind(Publisher.QUEUE_NAME1,Publisher.EXCHANGE_NAME,"");
        channel.queueBind(Publisher.QUEUE_NAME2,Publisher.EXCHANGE_NAME,"");


        //3. 监听queue
        DefaultConsumer callback = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消费者监听" + Publisher.QUEUE_NAME1 + "队列,获取到消息:" + new String(body));
                channel.basicAck(envelope.getDeliveryTag(),false);
            }
        };
        channel.basicConsume(Publisher.QUEUE_NAME1,callback);

        //4. stop
        System.out.println(Publisher.QUEUE_NAME1 + "开始监听");
        System.in.read();

        //5. c
        channel.close();
        conn.close();


    }


    @Test
    public void consume2() throws Exception, TimeoutException {
//1. c
        Connection conn = MQConnUtil.getConn();

        //2. c
        Channel channel = conn.createChannel();

        //3. 监听queue
        DefaultConsumer callback = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消费者监听" + Publisher.QUEUE_NAME2 + "队列,获取到消息:" + new String(body));
                channel.basicAck(envelope.getDeliveryTag(),false);
            }
        };
        channel.basicConsume(Publisher.QUEUE_NAME2,callback);

        //4. stop
        System.out.println(Publisher.QUEUE_NAME2 + "开始监听");
        System.in.read();

        //5. c
        channel.close();
        conn.close();


    }
}
